<!DOCTYPE html>












  


<html class="theme-next pisces use-motion" lang="zh-CN">
<head><meta name="generator" content="Hexo 3.8.0">
  <meta charset="UTF-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=2">
<meta name="theme-color" content="#222">
























<link rel="stylesheet" href="/lib/font-awesome/css/font-awesome.min.css?v=4.6.2">

<link rel="stylesheet" href="/css/main.css?v=7.1.0">


  <link rel="apple-touch-icon" sizes="180x180" href="/images/apple-touch-icon-next.png?v=7.1.0">


  <link rel="icon" type="image/png" sizes="32x32" href="/images/logo1.jpg?v=7.1.0">


  <link rel="icon" type="image/png" sizes="16x16" href="/images/logo1.jpg?v=7.1.0">


  <link rel="mask-icon" href="/images/logo.svg?v=7.1.0" color="#222">







<script id="hexo.configurations">
  var NexT = window.NexT || {};
  var CONFIG = {
    root: '/',
    scheme: 'Pisces',
    version: '7.1.0',
    sidebar: {"position":"left","display":"post","offset":12,"onmobile":false,"dimmer":false},
    back2top: true,
    back2top_sidebar: false,
    fancybox: false,
    fastclick: false,
    lazyload: false,
    tabs: true,
    motion: {"enable":true,"async":false,"transition":{"post_block":"fadeIn","post_header":"slideDownIn","post_body":"slideDownIn","coll_header":"slideLeftIn","sidebar":"slideUpIn"}},
    algolia: {
      applicationID: '',
      apiKey: '',
      indexName: '',
      hits: {"per_page":10},
      labels: {"input_placeholder":"Search for Posts","hits_empty":"We didn't find any results for the search: ${query}","hits_stats":"${hits} results found in ${time} ms"}
    }
  };
</script>


  




  <meta name="description" content="最近在做公司的数据平台项目，需要拿生产环境数据做数据分析。 当然数据分析这种事不能在生产环境的数据库上直接做，因为数据分析一般会有很复杂的SQL，直接在生产环境查询会对线上服务产生较大影响。 所以需要将生产环境数据同步到数据仓库（我们使用的是Kudu，关系型数据库，对这种mysql数据有很好的兼容性）">
<meta name="keywords" content="数据同步">
<meta property="og:type" content="article">
<meta property="og:title" content="数据平台底层链路搭建">
<meta property="og:url" content="https://luoo.oschina.io/2018/04/21/tech/数据平台底层链路搭建/index.html">
<meta property="og:site_name" content="从你的全世界路过">
<meta property="og:description" content="最近在做公司的数据平台项目，需要拿生产环境数据做数据分析。 当然数据分析这种事不能在生产环境的数据库上直接做，因为数据分析一般会有很复杂的SQL，直接在生产环境查询会对线上服务产生较大影响。 所以需要将生产环境数据同步到数据仓库（我们使用的是Kudu，关系型数据库，对这种mysql数据有很好的兼容性）">
<meta property="og:locale" content="zh-CN">
<meta property="og:image" content="https://luoo.oschina.io/images/data-link/data-link.jpg">
<meta property="og:updated_time" content="2019-04-14T12:28:00.758Z">
<meta name="twitter:card" content="summary">
<meta name="twitter:title" content="数据平台底层链路搭建">
<meta name="twitter:description" content="最近在做公司的数据平台项目，需要拿生产环境数据做数据分析。 当然数据分析这种事不能在生产环境的数据库上直接做，因为数据分析一般会有很复杂的SQL，直接在生产环境查询会对线上服务产生较大影响。 所以需要将生产环境数据同步到数据仓库（我们使用的是Kudu，关系型数据库，对这种mysql数据有很好的兼容性）">
<meta name="twitter:image" content="https://luoo.oschina.io/images/data-link/data-link.jpg">



  <link rel="alternate" href="/atom.xml" title="从你的全世界路过" type="application/atom+xml">



  
  
  <link rel="canonical" href="https://luoo.oschina.io/2018/04/21/tech/数据平台底层链路搭建/">



<script id="page.configurations">
  CONFIG.page = {
    sidebar: "",
  };
</script>

  <title>数据平台底层链路搭建 | 从你的全世界路过</title>
  












  <noscript>
  <style>
  .use-motion .motion-element,
  .use-motion .brand,
  .use-motion .menu-item,
  .sidebar-inner,
  .use-motion .post-block,
  .use-motion .pagination,
  .use-motion .comments,
  .use-motion .post-header,
  .use-motion .post-body,
  .use-motion .collection-title { opacity: initial; }

  .use-motion .logo,
  .use-motion .site-title,
  .use-motion .site-subtitle {
    opacity: initial;
    top: initial;
  }

  .use-motion .logo-line-before i { left: initial; }
  .use-motion .logo-line-after i { right: initial; }
  </style>
</noscript>

</head>

<body itemscope itemtype="http://schema.org/WebPage" lang="zh-CN">

  
  
    
  

  <div class="container sidebar-position-left page-post-detail">
    <div class="headband"></div>

    <header id="header" class="header" itemscope itemtype="http://schema.org/WPHeader">
      <div class="header-inner"><div class="site-brand-wrapper">
  <div class="site-meta">
    

    <div class="custom-logo-site-title">
      <a href="/" class="brand" rel="start">
        <span class="logo-line-before"><i></i></span>
        <span class="site-title">从你的全世界路过</span>
        <span class="logo-line-after"><i></i></span>
      </a>
    </div>
    
      
        <p class="site-subtitle">心智渐开，世事渐杂</p>
      
    
    
  </div>

  <div class="site-nav-toggle">
    <button aria-label="切换导航栏">
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
    </button>
  </div>
</div>



<nav class="site-nav">
  
    <ul id="menu" class="menu">
      
        
        
        
          
          <li class="menu-item menu-item-home">

    
    
    
      
    

    

    <a href="/" rel="section"><i class="menu-item-icon fa fa-fw fa-home"></i> <br>首页</a>

  </li>
        
        
        
          
          <li class="menu-item menu-item-categories">

    
    
    
      
    

    

    <a href="/categories/" rel="section"><i class="menu-item-icon fa fa-fw fa-th"></i> <br>分类</a>

  </li>
        
        
        
          
          <li class="menu-item menu-item-tags">

    
    
    
      
    

    

    <a href="/tags/" rel="section"><i class="menu-item-icon fa fa-fw fa-tags"></i> <br>标签</a>

  </li>
        
        
        
          
          <li class="menu-item menu-item-archives">

    
    
    
      
    

    

    <a href="/archives/" rel="section"><i class="menu-item-icon fa fa-fw fa-archive"></i> <br>归档</a>

  </li>
        
        
        
          
          <li class="menu-item menu-item-schedule">

    
    
    
      
    

    

    <a href="/schedule/" rel="section"><i class="menu-item-icon fa fa-fw fa-calendar"></i> <br>计划</a>

  </li>
        
        
        
          
          <li class="menu-item menu-item-you">

    
    
    
      
    

    

    <a href="/you/" rel="section"><i class="menu-item-icon fa fa-fw fa-heartbeat"></i> <br>知遇</a>

  </li>
        
        
        
          
          <li class="menu-item menu-item-about">

    
    
    
      
    

    

    <a href="/about/" rel="section"><i class="menu-item-icon fa fa-fw fa-user"></i> <br>关于</a>

  </li>

      
      
    </ul>
  

  

  
</nav>



  



</div>
    </header>

    


    <main id="main" class="main">
      <div class="main-inner">
        <div class="content-wrap">
          
            

          
          <div id="content" class="content">
            

  <div id="posts" class="posts-expand">
    

  

  
  
  

  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://luoo.oschina.io/2018/04/21/tech/数据平台底层链路搭建/">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="方林">
      <meta itemprop="description" content="愿你出走半生，归来仍是少年">
      <meta itemprop="image" content="/images/logo1.jpg">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="从你的全世界路过">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">数据平台底层链路搭建

              
            
          </h1>
        

        <div class="post-meta">
          <span class="post-time">

            
            
            

            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              

              
                
              

              <time title="创建时间：2018-04-21 14:20:25" itemprop="dateCreated datePublished" datetime="2018-04-21T14:20:25+08:00">2018-04-21</time>
            

            
          </span>

          
            <span class="post-category">
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing"><a href="/categories/编程技术/" itemprop="url" rel="index"><span itemprop="name">编程技术</span></a></span>

                
                
              
            </span>
          

          
            
            
              
              <span class="post-comments-count">
                <span class="post-meta-divider">|</span>
                <span class="post-meta-item-icon">
                  <i class="fa fa-comment-o"></i>
                </span>
            
                <span class="post-meta-item-text">评论数：</span>
                <a href="/2018/04/21/tech/数据平台底层链路搭建/#comments" itemprop="discussionUrl">
                  <span class="post-comments-count valine-comment-count" data-xid="/2018/04/21/tech/数据平台底层链路搭建/" itemprop="commentCount"></span>
                </a>
              </span>
            
          

          
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        <p>最近在做公司的数据平台项目，需要拿生产环境数据做数据分析。</p>
<p>当然数据分析这种事不能在生产环境的数据库上直接做，因为数据分析一般会有很复杂的SQL，直接在生产环境查询会对线上服务产生较大影响。</p>
<p>所以需要将生产环境数据同步到数据仓库（我们使用的是Kudu，关系型数据库，对这种mysql数据有很好的兼容性）</p>
<a id="more"></a>
<h2 id="引言"><a href="#引言" class="headerlink" title="引言"></a>引言</h2><p>经过调研，我们选取了 Mysql -&gt; Canal -&gt; Kafka -&gt; Spark Streaming -&gt; (Kudu + Impala) 的设计方案，上层由Ad-hoc封装，提供http协议的接口访问。</p>
<p><img src="/images/data-link/data-link.jpg" alt="datalink"></p>
<p>下面分别介绍下这几个组件：</p>
<h3 id="Mysql"><a href="#Mysql" class="headerlink" title="Mysql"></a>Mysql</h3><p>Mysql就不用多说了，应用最广泛的关系型数据库。</p>
<h3 id="Canal"><a href="#Canal" class="headerlink" title="Canal"></a>Canal</h3><p>Canal 是有alibaba开源的跨机房数据同步工具。它的原理是伪装mysql的slave机接收来自master的binlog，解析数据并同步给下游。</p>
<h3 id="Kafka"><a href="#Kafka" class="headerlink" title="Kafka"></a>Kafka</h3><p>消息中间件，在这里除了做消息中间件的作用外，还有缓存和回溯数据的作用。</p>
<h3 id="Spark-Streaming"><a href="#Spark-Streaming" class="headerlink" title="Spark Streaming"></a>Spark Streaming</h3><p>流式数据计算框架，用于数据转换、聚合和计算。</p>
<h3 id="Kudu"><a href="#Kudu" class="headerlink" title="Kudu"></a>Kudu</h3><p>数据仓库。</p>
<h3 id="Impala"><a href="#Impala" class="headerlink" title="Impala"></a>Impala</h3><p>Kudu的查询工具。</p>
<h2 id="实践"><a href="#实践" class="headerlink" title="实践"></a>实践</h2><h3 id="安装配置mysql"><a href="#安装配置mysql" class="headerlink" title="安装配置mysql"></a>安装配置mysql</h3><ol>
<li><strong>安装mysql</strong><br>mysql的安装就不详细介绍了，网上很多教程。<br><em>请注意的是Canal对于mysql的版本是有要求的</em></li>
<li><p><strong>配置</strong><br>在/etc/my.cnf中添加如下配置，重启服务</p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line">log-bin=mysql-bin #添加这一行就ok  </span><br><span class="line">binlog-format=ROW #选择row模式  </span><br><span class="line">server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复</span><br></pre></td></tr></table></figure>
</li>
<li><p><strong>授权用户</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line">CREATE USER canal IDENTIFIED BY &apos;canal&apos;;    </span><br><span class="line">GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO &apos;canal&apos;@&apos;%&apos;;  </span><br><span class="line">GRANT ALL PRIVILEGES ON *.* TO &apos;canal&apos;@&apos;%&apos; ;  </span><br><span class="line">FLUSH PRIVILEGES;</span><br></pre></td></tr></table></figure>
</li>
</ol>
<h3 id="Canal-1"><a href="#Canal-1" class="headerlink" title="Canal"></a>Canal</h3><p>Canal分为Server和Client，在这里我们只配置服务端，后面介绍怎么编写客户端把数据同步到Kafka</p>
<ol>
<li><strong>安装jdk</strong><br>统一使用JDK1.8</li>
<li><p><strong>下载Canal Server</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line">https://github.com/alibaba/canal/releases </span><br><span class="line"></span><br><span class="line">下载release版或者自己编译</span><br><span class="line"></span><br><span class="line">cd canal</span><br><span class="line">git checkout -b branch tag</span><br><span class="line">mvn clean install -Dmaven.test.skip -Denv=release</span><br><span class="line">编译完成后，会在根目录下产生target/canal.deployer-$version.tar.gz</span><br><span class="line"></span><br><span class="line">tar zxvf canal.deployer-$version.tar.gz</span><br></pre></td></tr></table></figure>
</li>
<li><p><strong>配置canal-server</strong>  <code>vi conf/canal.properties</code></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line">####   需要关注的参数</span><br><span class="line">canal.port= 11111</span><br><span class="line">canal.destinations= example</span><br><span class="line"></span><br><span class="line"># position info，需要改成自己的数据库信息</span><br><span class="line">canal.instance.master.address = 127.0.0.1:3306</span><br><span class="line">canal.instance.master.journal.name =</span><br><span class="line">canal.instance.master.position =</span><br><span class="line">canal.instance.master.timestamp =</span><br><span class="line"></span><br><span class="line"># username/password，需要改成自己的数据库信息  </span><br><span class="line">canal.instance.dbUsername = root</span><br><span class="line">canal.instance.dbPassword = 123456</span><br><span class="line">canal.instance.defaultDatabaseName = datamip  </span><br><span class="line">canal.instance.connectionCharset = UTF-8  </span><br><span class="line"></span><br><span class="line"># table regex  </span><br><span class="line">canal.instance.filter.regex = .*\\..*</span><br></pre></td></tr></table></figure>
</li>
<li><p><strong>配置instance</strong> <code>vi conf/example/instance.properties</code></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line">####   需要关注的参数</span><br><span class="line">canal.instance.master.address = 192.168.23.1:3306</span><br><span class="line">canal.instance.dbUsername = canal</span><br><span class="line">canal.instance.dbPassword = canal</span><br><span class="line">canal.instance.defaultDatabaseName =datamip</span><br><span class="line">canal.instance.connectionCharset = UTF-8</span><br><span class="line"># table regex</span><br><span class="line">canal.instance.filter.regex = .*\\..*</span><br><span class="line"># table black regex</span><br><span class="line">canal.instance.filter.black.regex =</span><br></pre></td></tr></table></figure>
</li>
<li><p><strong>运行/停止</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">bin/startup.sh</span><br><span class="line">bin/stop.sh</span><br></pre></td></tr></table></figure>
</li>
</ol>
<h3 id="Kafka-1"><a href="#Kafka-1" class="headerlink" title="Kafka"></a>Kafka</h3><ol>
<li><p><strong>下载</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line">http://mirrors.hust.edu.cn/apache/kafka/1.0.1/kafka_2.11-1.0.1.tgz</span><br><span class="line">tar -xzf kafka_2.11-1.0.1.tgz</span><br><span class="line">cd kafka_2.11-1.0.1</span><br></pre></td></tr></table></figure>
</li>
<li><p><strong>运行</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">bin/zookeeper-server-start.sh config/zookeeper.properties</span><br><span class="line">bin/kafka-server-start.sh config/server.properties</span><br></pre></td></tr></table></figure>
</li>
</ol>
<h3 id="Canal-gt-Kafka"><a href="#Canal-gt-Kafka" class="headerlink" title="Canal-&gt;Kafka"></a>Canal-&gt;Kafka</h3><ol>
<li><p><strong>Canal Client</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br></pre></td><td class="code"><pre><span class="line">CanalConnector connector = CanalConnectors.newSingleConnector(</span><br><span class="line">        new InetSocketAddress(cfg.canal_server(), cfg.canal_port()),</span><br><span class="line">        cfg.canal_destination(),</span><br><span class="line">        cfg.canal_username(),</span><br><span class="line">        cfg.canal_password()</span><br><span class="line">);</span><br><span class="line"></span><br><span class="line">regex = cfg.canal_table_filter();</span><br><span class="line">dbs = cfg.canal_db_filter().split(&quot;,&quot;);</span><br><span class="line"></span><br><span class="line">String prefix = cfg.canal_subscribe_prefix();</span><br><span class="line"></span><br><span class="line">int batchSize = 1000;</span><br><span class="line">try &#123;</span><br><span class="line">    connector.connect();</span><br><span class="line">    connector.subscribe(prefix + &quot;.*\\.*&quot;);</span><br><span class="line">    connector.rollback();</span><br><span class="line">    log.info(&quot;Client Started ...&quot;);</span><br><span class="line">    while (true) &#123;</span><br><span class="line">        // 获取指定数量的数据</span><br><span class="line">        Message message = connector.getWithoutAck(batchSize);</span><br><span class="line">        long batchId = message.getId();</span><br><span class="line">        int size = message.getEntries().size();</span><br><span class="line">        if (batchId == -1 || size == 0) &#123;</span><br><span class="line">            try &#123;</span><br><span class="line">                Thread.sleep(200);</span><br><span class="line">                continue;</span><br><span class="line">            &#125;catch (InterruptedException e) &#123;</span><br><span class="line">                log.error(&quot;线程调度错误&quot;, e);</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        try &#123;</span><br><span class="line">            log.info(&quot;batchId:&quot; + batchId + &quot; Size:&quot; +size);</span><br><span class="line">            doExecute(message);</span><br><span class="line">            // 提交确认</span><br><span class="line">            connector.ack(batchId);</span><br><span class="line">        &#125; catch (Exception e) &#123;</span><br><span class="line">            // 处理失败, 回滚数据</span><br><span class="line">            log.error(&quot;处理数据失败&quot;, message, e);</span><br><span class="line">            connector.rollback(batchId);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">&#125; catch (Exception e) &#123;</span><br><span class="line">    log.error(&quot;MAIN LOOP CATCH EXCEPTION : &quot; + e);</span><br><span class="line">&#125; finally &#123;</span><br><span class="line">    connector.disconnect();</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
</li>
<li><p><strong>Kafka Producer</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br></pre></td><td class="code"><pre><span class="line">/**</span><br><span class="line">* 发送消息</span><br><span class="line">* @param topic topicName</span><br><span class="line">* @param key key</span><br><span class="line">* @param data data</span><br><span class="line">*/</span><br><span class="line">public static void sendMsg(String topic, String key, String data)&#123;</span><br><span class="line">    try &#123;</span><br><span class="line">        Producer producer = createProducer();</span><br><span class="line">        producer.send(new ProducerRecord&lt;&gt;(topic, key, data));</span><br><span class="line">    &#125; catch (Exception e) &#123;</span><br><span class="line">        logger.error(&quot;SEND ERROR &quot; + e.getMessage());</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">/**</span><br><span class="line">* @return kafka_producer</span><br><span class="line">*/</span><br><span class="line">private static KafkaProducer&lt;String,String&gt; createProducer()&#123;</span><br><span class="line">    if (client != null) &#123;</span><br><span class="line">        return client;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    ServerConfig cfg = ConfigFactory.create(ServerConfig.class);</span><br><span class="line">    Properties properties = new Properties();</span><br><span class="line">    properties.put(&quot;bootstrap.servers&quot;, cfg.kafka_bootstrap_servers());</span><br><span class="line">    properties.put(&quot;acks&quot;, cfg.kafka_acks());</span><br><span class="line">    properties.put(&quot;retries&quot;, cfg.kafka_retries());</span><br><span class="line">    properties.put(&quot;batch.size&quot;, cfg.kafka_batch_size());</span><br><span class="line">    properties.put(&quot;linger.ms&quot;, cfg.kafka_linger_ms());</span><br><span class="line">    properties.put(&quot;buffer.memory&quot;, cfg.kafka_buffer_memory());</span><br><span class="line">    properties.put(&quot;key.serializer&quot;, cfg.kafka_key_serializer());</span><br><span class="line">    properties.put(&quot;value.serializer&quot;, cfg.kafka_value_serializer());</span><br><span class="line">    client = new KafkaProducer&lt;&gt;(properties);</span><br><span class="line">    return client;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
</li>
</ol>
<h3 id="Spark-Streaming-1"><a href="#Spark-Streaming-1" class="headerlink" title="Spark Streaming"></a>Spark Streaming</h3><ol>
<li><p><strong>Kafka -&gt; spark Streaming -&gt; Kudu</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br></pre></td><td class="code"><pre><span class="line">object StreamingToKuduUseDirect extends Serializable &#123;</span><br><span class="line"></span><br><span class="line">    def main(args: Array[String]): Unit = &#123;</span><br><span class="line">        </span><br><span class="line">        //1.获取输入参数与定义全局变量,主要两部分，一个brokers地址，一个topics列表，至于其他优化参数后续再加</span><br><span class="line">        //TODO: 注意：后面可以改为properties的方式来获取参数</span><br><span class="line">        val Array(brokers, topic, masterList, kuduTableName, appName, dataProcessingMode, groupid, checkpoint_path) = args</span><br><span class="line">        val sctime = 10</span><br><span class="line"></span><br><span class="line">        //2.配置spark环境以及kudu环境</span><br><span class="line">        val spark = SparkSession.builder.appName(appName).enableHiveSupport().getOrCreate()</span><br><span class="line">        val ssc = new StreamingContext(spark.sparkContext, Seconds(sctime))</span><br><span class="line"></span><br><span class="line">        ssc.checkpoint(checkpoint_path)</span><br><span class="line"></span><br><span class="line">        val kuduContext = new KuduContext(masterList, spark.sparkContext)</span><br><span class="line"></span><br><span class="line">        //3.配置创建kafka输入流的所需要的参数，注意这里可以加入一些优化参数</span><br><span class="line">        val kafkaParams = Map[String, Object](</span><br><span class="line">            &quot;bootstrap.servers&quot; -&gt; brokers,</span><br><span class="line">            &quot;key.deserializer&quot; -&gt; classOf[StringDeserializer],</span><br><span class="line">            &quot;value.deserializer&quot; -&gt; classOf[StringDeserializer],</span><br><span class="line">            &quot;group.id&quot; -&gt; groupid,</span><br><span class="line">            &quot;auto.offset.reset&quot; -&gt; &quot;latest&quot;,</span><br><span class="line">            &quot;enable.auto.commit&quot; -&gt; (true: java.lang.Boolean)</span><br><span class="line">        )</span><br><span class="line"></span><br><span class="line">        val topics = Array(topic)</span><br><span class="line"></span><br><span class="line">        //4.创建kafka输入流0.10</span><br><span class="line">        val stream = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams))</span><br><span class="line"></span><br><span class="line">        stream.foreachRDD &#123; rdd =&gt;</span><br><span class="line"></span><br><span class="line">            //获取所有的value</span><br><span class="line">            val messageRDD = rdd.map(record =&gt; (record.key(), record.value())).values</span><br><span class="line"></span><br><span class="line">            //Data Processing</span><br><span class="line">            val kuduDF = dataProcessing.Porcess(dataProcessingMode, messageRDD, spark)</span><br><span class="line"></span><br><span class="line">            //Write Kudu</span><br><span class="line">            kuduContext.upsertRows(kuduDF, kuduTableName)</span><br><span class="line"></span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        //6.Start the computation</span><br><span class="line">        ssc.start()</span><br><span class="line">        ssc.awaitTermination()</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
</li>
<li><p><strong>消息格式</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br></pre></td><td class="code"><pre><span class="line">Canal-&gt;Kafka 消息格式</span><br><span class="line"></span><br><span class="line">/**</span><br><span class="line"> *  消息片</span><br><span class="line"> */</span><br><span class="line">public class MessagePiece &#123;</span><br><span class="line"></span><br><span class="line">    /**</span><br><span class="line">     * 表名称</span><br><span class="line">     */</span><br><span class="line">    public String table;</span><br><span class="line"></span><br><span class="line">    /**</span><br><span class="line">     *  操作类型</span><br><span class="line">     */</span><br><span class="line">    public String type;</span><br><span class="line"></span><br><span class="line">    /**</span><br><span class="line">     *  修改后数据</span><br><span class="line">     */</span><br><span class="line">    public Object row_after;</span><br><span class="line"></span><br><span class="line">    /**</span><br><span class="line">     * 修改前数据</span><br><span class="line">     */</span><br><span class="line">    public Object row_before;</span><br><span class="line"></span><br><span class="line">    /**</span><br><span class="line">     * constructor</span><br><span class="line">     */</span><br><span class="line">    public MessagePiece()&#123;</span><br><span class="line">        this.type = &quot;&quot;;</span><br><span class="line">        this.table = &quot;&quot;;</span><br><span class="line">        this.row_after = &quot;&quot;;</span><br><span class="line">        this.row_before = &quot;&quot;;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
</li>
<li><p><strong>Data Processing</strong></p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br></pre></td><td class="code"><pre><span class="line">Spark Streaming 数据抽取逻辑</span><br><span class="line"></span><br><span class="line">/**</span><br><span class="line">* 执行函数</span><br><span class="line">* @param javaPairDStream 数据流</span><br><span class="line">* @param broadcast 广播Session</span><br><span class="line">* @return</span><br><span class="line">*/</span><br><span class="line">override def execute(javaPairDStream: JavaPairDStream[String, String], broadcast: Broadcast[SparkSession]): JavaDStream[KuduTableEvent] = &#123;</span><br><span class="line">    try &#123;</span><br><span class="line">        javaPairDStream.dstream.transform((transformFunc: RDD[(String, String)]) =&gt; &#123;</span><br><span class="line">            transformFunc.map(msg =&gt; &#123;</span><br><span class="line">                implicit val formats: DefaultFormats.type = DefaultFormats</span><br><span class="line">                val item = parse(msg._2) </span><br><span class="line">                val op_type = (item \ &quot;type&quot;).extract[String]</span><br><span class="line">                val table = (item \ &quot;table&quot;).extract[String]</span><br><span class="line">                op_type.trim match &#123;</span><br><span class="line">                    case &quot;INSERT&quot; =&gt; new KuduTableEvent(table, getDataFrame(broadcast, item, &quot;row_after&quot;, table), null)</span><br><span class="line">                    case &quot;UPDATE&quot; =&gt; new KuduTableEvent(table, getDataFrame(broadcast, item, &quot;row_after&quot;, table), null)</span><br><span class="line">                    case &quot;DELETE&quot; =&gt; new KuduTableEvent(table, null, getDataFrame(broadcast, item, &quot;row_before&quot;, table))</span><br><span class="line">                    case _ =&gt; new KuduTableEvent(&quot;&quot;, null, null)</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;)</span><br><span class="line">        &#125;)</span><br><span class="line">    &#125; catch &#123;</span><br><span class="line">        case ex: Exception =&gt;</span><br><span class="line">            log.error(&quot;Handle PairDStream error : &quot;, ex)</span><br><span class="line">            null</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
</li>
</ol>
<h2 id="总结"><a href="#总结" class="headerlink" title="总结"></a>总结</h2><p>在整条链路的搭建的过程中，踩过很多坑，下面列出几个关键的：</p>
<h3 id="关于事务"><a href="#关于事务" class="headerlink" title="关于事务"></a>关于事务</h3><p>由于消息体只是一条消息，只能包含一条数据的增删改；<br>从这个层面上看，是无法支持事务的；<br>但是从另一个角度想，如果能保证所有消息消费跟原始生产时序一致，且每条消息消费一定成功；<br>那么事务的有无，就没有意义了；<br>即使中间有那个地方crash了，也可以通过Kafka回溯数据；<br>所以，事务在整个流程中可以不用考虑。</p>
<h3 id="Kafka的Key"><a href="#Kafka的Key" class="headerlink" title="Kafka的Key"></a>Kafka的Key</h3><p>我们都知道Kafka为了提升IO，可以将一个topic分为多个partition的；<br>如果一个topic有多个partition，下游消费的时候就无法保证整体时序了；<br>我们开始想过几个方案：<br>a) 一个数据库一个topic，一个topic可以有多个partition，通过表名hash保证一个表的增量数据只会进入一个partition；<br>b) 一张表一个topic，一个topic只有一个parittion；</p>
<p>这两种方案有个共同的特点：<br>都不能保证整体消费时序，但是都能保证对于每张表的增量消息消费时时序的，这一点很关键，足以保证数据的正确性了。<br>这两种方案都是对表级别的数据进行隔离。</p>
<p>但是他们都有缺点：<br>第一种方案，由于每张表的数据写操作频率不同，topic中的消息在partition中分布会很不均匀；<br>第二种方案，要维护的topic太多，而且topic之间的消息量也可能差别很大。</p>
<p>推荐方案：<br>使用业务数据的ID作为Kafka的Key（利用Kafka的Key特性，一个topic中相同Key的数据必定会进入同一个partition）。<br>这种方案是对行级别的数据进行隔离，保证对每行数据的修改的时序，也能保证数据的正确性。</p>
<h3 id="数据导入"><a href="#数据导入" class="headerlink" title="数据导入"></a>数据导入</h3><p>整条链路是一条增量信息链路，那么对于初始数据和表结构，我们采用程序要导入，这里面遇到了几个坑：<br>a)  mysql的数据字段和Kudu不完全兼容，特别是经过程序插入的时候有很多类型限制，譬如时间戳，因此需要做一些类型映射和转换，参加如下代码；<br>Schema Class<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br></pre></td><td class="code"><pre><span class="line">switch (type) &#123;</span><br><span class="line">    case &quot;INT&quot; :</span><br><span class="line">    case &quot;TINYINT&quot; :</span><br><span class="line">    case &quot;SMALLINT&quot;:</span><br><span class="line">    case &quot;TIMESTAMP&quot; :</span><br><span class="line">    case &quot;INT UNSIGNED&quot; :</span><br><span class="line">    case &quot;TINYINT UNSIGNED&quot; :</span><br><span class="line">    case &quot;SMALLINT UNSIGNED&quot;:</span><br><span class="line">        classStr += &quot;,\n\t&quot; + realName + &quot;: Int&quot;;</span><br><span class="line">        break;</span><br><span class="line">    case &quot;BIGINT&quot; :</span><br><span class="line">    case &quot;BIGINT UNSIGNED&quot; :</span><br><span class="line">        classStr += &quot;,\n\t&quot; + realName + &quot;: Long&quot;;</span><br><span class="line">        break;</span><br><span class="line">    case &quot;VARCHAR&quot; :</span><br><span class="line">    case &quot;CHAR&quot; :</span><br><span class="line">        classStr += &quot;,\n\t&quot; + realName + &quot;: String&quot;;</span><br><span class="line">        break;</span><br><span class="line">    case &quot;DECIMAL&quot; :</span><br><span class="line">    case &quot;DECIMAL UNSIGNED&quot; :</span><br><span class="line">        classStr += &quot;,\n\t&quot; + realName + &quot;: Double&quot;;</span><br><span class="line">        break;</span><br><span class="line">    case &quot;BOOL&quot; :</span><br><span class="line">        classStr += &quot;,\n\t&quot; + realName + &quot;: Boolean&quot;;</span><br><span class="line">        break;</span><br><span class="line">    default:</span><br><span class="line">        System.out.println(&quot;Error key&quot; + name + &quot;:&quot; + type);</span><br><span class="line">        break;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p>Insert Mode<br><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br></pre></td><td class="code"><pre><span class="line">switch (type) &#123;</span><br><span class="line">    case &quot;TINYINT&quot; :</span><br><span class="line">    case &quot;TINYINT UNSIGNED&quot; :</span><br><span class="line">    case &quot;INT&quot; :</span><br><span class="line">    case &quot;SMALLINT&quot;:</span><br><span class="line">    case &quot;INT UNSIGNED&quot; :</span><br><span class="line">    case &quot;SMALLINT UNSIGNED&quot;:</span><br><span class="line">        realType = value.getClass().getSimpleName();</span><br><span class="line">        if (realType.equals(&quot;Boolean&quot;)) &#123;</span><br><span class="line">            row.addInt(realKey, (boolean) value ? 1 : 0);</span><br><span class="line">        &#125; else if (realType.equals(&quot;Long&quot;)) &#123;</span><br><span class="line">            row.addInt(realKey, ((Long) value).intValue());</span><br><span class="line">        &#125; else &#123;</span><br><span class="line">            row.addInt(realKey, (int) value);</span><br><span class="line">        &#125;</span><br><span class="line">        break;</span><br><span class="line">    case &quot;BIGINT&quot; :</span><br><span class="line">    case &quot;BIGINT UNSIGNED&quot; :</span><br><span class="line">        realType = value.getClass().getSimpleName();</span><br><span class="line">        if (realType.equals(&quot;BigInteger&quot;)) &#123;</span><br><span class="line">            row.addLong(realKey, ((BigInteger) value).longValue());</span><br><span class="line">        &#125; else &#123;</span><br><span class="line">            row.addLong(realKey, (Long) value);</span><br><span class="line">        &#125;</span><br><span class="line">        break;</span><br><span class="line">    case &quot;CHAR&quot; :</span><br><span class="line">    case &quot;VARCHAR&quot; :</span><br><span class="line">        row.addString(realKey, (String) value);</span><br><span class="line">        break;</span><br><span class="line">    case &quot;DECIMAL&quot; :</span><br><span class="line">    case &quot;DECIMAL UNSIGNED&quot; :</span><br><span class="line">        if (value.getClass().getSimpleName().equals(&quot;BigDecimal&quot;)) &#123;</span><br><span class="line">            row.addDouble(realKey, ((BigDecimal) value).doubleValue());</span><br><span class="line">        &#125; else &#123;</span><br><span class="line">            row.addDouble(realKey, (Double) value);</span><br><span class="line">        &#125;</span><br><span class="line">        break;</span><br><span class="line">    case &quot;TIMESTAMP&quot; :</span><br><span class="line">        int t = (int)((java.sql.Timestamp)value).toInstant().getEpochSecond();</span><br><span class="line">        row.addInt(realKey, t);</span><br><span class="line">        break;</span><br><span class="line">    case &quot;BOOL&quot; :</span><br><span class="line">        row.addBoolean(realKey, (boolean) value);</span><br><span class="line">        break;</span><br><span class="line">    default :</span><br><span class="line">        logger.error(&quot;UNKNOWN type : &quot; + type);</span><br><span class="line">        break;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p>b)  mysql中tinyint(1)类型，这简直就是个坑。jdbc在读取mysql中数据时，tinyint(1)的类型数据的值会直接被转成bool值，解决方案在mysql的连接串后添加<code>?tinyInt1isBit=false</code>；</p>
<h3 id="Spark消费Kafka"><a href="#Spark消费Kafka" class="headerlink" title="Spark消费Kafka"></a>Spark消费Kafka</h3><p>关于spark消费Kafka存在一个问题，spark每次从kafka中拿取数据是从每个partition中拿一个批次，一个批次中可能会包含很多条数据，这些数据在取出来的时候是有序的，但是在消费时每条消息作为一个任务被分发到spark集群的各个executor，每个executor什么时候执行完成都是不定的，因此实际的消费顺序是无法保证的。</p>
<p>说到这个问题，我们在上线的时候还捅了一个很严重的线上事故：<br>最开始的时候，我们不知道spark可以控制消费kafka的速率，就没做任何消费配置，这样直接接上kafka集群消费。<br>我们的spark集群有100多台机器，而kafka集群只有三台broker，在上线前我们已经把上半段链路打通上线了，kafka中缓存了近7天的所有消息，应该有几百万条数据。<br>此时我们直接上spark集群从最早开始消费kafka，设置的event周期是500ms，就是每500ms从kafka取一次数据。<br>在spark任务执行了不到半分钟的时候，我们看到kafka的out网络流量，从初始的5k~6k一下飙升到30M~35M，<br>紧接着在kafka monitor中看到kafka集群中部分topic的少了一个broker，<br>当时我有所警觉，但是由于生产环境kafka不属于我们部门维护，加上个人对kafka集群不是特别熟悉，就截图监控问了一下基础架构部的同事，<br>在还没有得到答复之前，就开始陆陆续续收到各个业务监控的MQ发送和消费失败报警，<br>想到我们的MQ和基于kafka的封装的，而且我现在所使用的kafka集群正是MQ所使用的那个，我开始怀疑是kafka集群挂掉了，<br>此时基础架构部的人才开始回复我kafka有问题，然后我问是不是刚才一波消费高峰导致的，就没人理我了😂（都去忙着处理线上报警了）<br>后来事故总结，首要原因，kafka集群的内存设置太小，导致内存溢出，集群呈假死状态；第二，spark集群消费没做速率限制，每次从一个partition拿了近万条数据，10个partition加起来就几十万条数据了。</p>
<p>上线之后，我们又发现一个特别坑的问题：<br>是关于unsigned的，在上线初期我们并没考虑到unsigned会到来怎样的后果，<br>因为不管有没有unsigned，改字段实际存储的空间大小是一样的，不管怎样都能存储进去，直到真正上线后，遇到问题才发现。<br>mysql中关于整形的数据都可以设置unsigned标识，而java和kudu却没有unsigned标识，<br>当数据通过明文网络传输后，从mysql中取到的unsigned int数据，较大的数据在java和kudu中解析就变成了bigint，数据插入会失败，<br>最开始想到的简单粗暴方法是把所有类型都转成string，这样存储什么都不是问题了，但是SUM聚合函数才SQL中就无法使用了（行不通），<br>所以针对整形的类型，不管变成int还是bigint，类型最终还得是整形，因此只能做类型上升，unsigned short转int，unsigned int转bigint，<br>但是unsigned bigint就很尴尬了，没有类型可上升，如果非要上升只能是string了，但是string又没法做sum运算，<br>后来想到，我们业务代码都是用php写的，php中也没有unsigned bigint这个东西，php整形能表达的最大数据也就是bigint的最大数据，<br>所以数据库中unsigned bigint最大的数据也只可能是bigint最大的数据，如果有比这个大的业务代码肯定会抛异常了，<br>因此unsigned bigint这个类型可以考虑不用上升了，保持原状即可。</p>

      
    </div>

    

    
    
    

    

    
      
    
    

    

    <footer class="post-footer">
      
        <div class="post-tags">
          
            <a href="/tags/数据同步/" rel="tag"># 数据同步</a>
          
        </div>
      

      
      
      

      
        <div class="post-nav">
          <div class="post-nav-next post-nav-item">
            
              <a href="/2018/03/06/tech/SQL语句Tips/" rel="next" title="SQL语句Tips">
                <i class="fa fa-chevron-left"></i> SQL语句Tips
              </a>
            
          </div>

          <span class="post-nav-divider"></span>

          <div class="post-nav-prev post-nav-item">
            
              <a href="/2018/09/05/tech/数据仓库迁移/" rel="prev" title="数据仓库迁移">
                数据仓库迁移 <i class="fa fa-chevron-right"></i>
              </a>
            
          </div>
        </div>
      

      
      
    </footer>
  </div>
  
  
  
  </article>


  </div>


          </div>
          

  
    <div class="comments" id="comments">
    </div>

  



        </div>
        
          
  
  <div class="sidebar-toggle">
    <div class="sidebar-toggle-line-wrap">
      <span class="sidebar-toggle-line sidebar-toggle-line-first"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-middle"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-last"></span>
    </div>
  </div>

  <aside id="sidebar" class="sidebar">
    <div class="sidebar-inner">

      

      
        <ul class="sidebar-nav motion-element">
          <li class="sidebar-nav-toc sidebar-nav-active" data-target="post-toc-wrap">
            文章目录
          </li>
          <li class="sidebar-nav-overview" data-target="site-overview-wrap">
            站点概览
          </li>
        </ul>
      

      <div class="site-overview-wrap sidebar-panel">
        <div class="site-overview">
          <div class="site-author motion-element" itemprop="author" itemscope itemtype="http://schema.org/Person">
            
              <img class="site-author-image" itemprop="image" src="/images/logo1.jpg" alt="方林">
            
              <p class="site-author-name" itemprop="name">方林</p>
              <div class="site-description motion-element" itemprop="description">愿你出走半生，归来仍是少年</div>
          </div>

          
            <nav class="site-state motion-element">
              
                <div class="site-state-item site-state-posts">
                
                  <a href="/archives/">
                
                    <span class="site-state-item-count">27</span>
                    <span class="site-state-item-name">日志</span>
                  </a>
                </div>
              

              
                
                
                <div class="site-state-item site-state-categories">
                  
                    
                      <a href="/categories/">
                    
                  
                    
                    
                      
                    
                      
                    
                      
                    
                    <span class="site-state-item-count">3</span>
                    <span class="site-state-item-name">分类</span>
                  </a>
                </div>
              

              
                
                
                <div class="site-state-item site-state-tags">
                  
                    
                      <a href="/tags/">
                    
                  
                    
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                      
                    
                    <span class="site-state-item-count">17</span>
                    <span class="site-state-item-name">标签</span>
                  </a>
                </div>
              
            </nav>
          

          
            <div class="feed-link motion-element">
              <a href="/atom.xml" rel="alternate">
                <i class="fa fa-rss"></i>
                RSS
              </a>
            </div>
          

          

          
            <div class="links-of-author motion-element">
              
                <span class="links-of-author-item">
                  
                  
                    
                  
                  
                    
                  
                  <a href="https://github.com/yourname" title="GitHub &rarr; https://github.com/yourname" rel="noopener" target="_blank"><i class="fa fa-fw fa-github"></i>GitHub</a>
                </span>
              
                <span class="links-of-author-item">
                  
                  
                    
                  
                  
                    
                  
                  <a href="mailto:sean_ps@163.com" title="E-Mail &rarr; mailto:sean_ps@163.com" rel="noopener" target="_blank"><i class="fa fa-fw fa-envelope"></i>E-Mail</a>
                </span>
              
            </div>
          

          

          
          
            <div class="links-of-blogroll motion-element links-of-blogroll-block">
              <div class="links-of-blogroll-title">
                <i class="fa  fa-fw fa-link"></i>
                Links
              </div>
              <ul class="links-of-blogroll-list">
                
                  <li class="links-of-blogroll-item">
                    <a href="http://www.luoo.net" title="http://www.luoo.net" rel="noopener" target="_blank">落网</a>
                  </li>
                
              </ul>
            </div>
          

          
            
          
          

        </div>
      </div>

      
      <!--noindex-->
        <div class="post-toc-wrap motion-element sidebar-panel sidebar-panel-active">
          <div class="post-toc">

            
            
            
            

            
              <div class="post-toc-content"><ol class="nav"><li class="nav-item nav-level-2"><a class="nav-link" href="#引言"><span class="nav-number">1.</span> <span class="nav-text">引言</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#Mysql"><span class="nav-number">1.1.</span> <span class="nav-text">Mysql</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Canal"><span class="nav-number">1.2.</span> <span class="nav-text">Canal</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Kafka"><span class="nav-number">1.3.</span> <span class="nav-text">Kafka</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Spark-Streaming"><span class="nav-number">1.4.</span> <span class="nav-text">Spark Streaming</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Kudu"><span class="nav-number">1.5.</span> <span class="nav-text">Kudu</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Impala"><span class="nav-number">1.6.</span> <span class="nav-text">Impala</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#实践"><span class="nav-number">2.</span> <span class="nav-text">实践</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#安装配置mysql"><span class="nav-number">2.1.</span> <span class="nav-text">安装配置mysql</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Canal-1"><span class="nav-number">2.2.</span> <span class="nav-text">Canal</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Kafka-1"><span class="nav-number">2.3.</span> <span class="nav-text">Kafka</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Canal-gt-Kafka"><span class="nav-number">2.4.</span> <span class="nav-text">Canal-&gt;Kafka</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Spark-Streaming-1"><span class="nav-number">2.5.</span> <span class="nav-text">Spark Streaming</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#总结"><span class="nav-number">3.</span> <span class="nav-text">总结</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#关于事务"><span class="nav-number">3.1.</span> <span class="nav-text">关于事务</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Kafka的Key"><span class="nav-number">3.2.</span> <span class="nav-text">Kafka的Key</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#数据导入"><span class="nav-number">3.3.</span> <span class="nav-text">数据导入</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Spark消费Kafka"><span class="nav-number">3.4.</span> <span class="nav-text">Spark消费Kafka</span></a></li></ol></li></ol></div>
            

          </div>
        </div>
      <!--/noindex-->
      

      

    </div>
  </aside>
  


        
      </div>
    </main>

    <footer id="footer" class="footer">
      <div class="footer-inner">
        <div class="copyright">&copy; <span itemprop="copyrightYear">2019</span>
  <span class="with-love" id="animate">
    <i class="fa fa-user"></i>
  </span>
  <span class="author" itemprop="copyrightHolder">fanglin</span>

  

  
</div>









        








        
      </div>
    </footer>

    
      <div class="back-to-top">
        <i class="fa fa-arrow-up"></i>
        
      </div>
    

    

    

    
  </div>

  

<script>
  if (Object.prototype.toString.call(window.Promise) !== '[object Function]') {
    window.Promise = null;
  }
</script>


























  
  <script src="/lib/jquery/index.js?v=2.1.3"></script>

  
  <script src="/lib/velocity/velocity.min.js?v=1.2.1"></script>

  
  <script src="/lib/velocity/velocity.ui.min.js?v=1.2.1"></script>


  


  <script src="/js/utils.js?v=7.1.0"></script>

  <script src="/js/motion.js?v=7.1.0"></script>



  
  


  <script src="/js/affix.js?v=7.1.0"></script>

  <script src="/js/schemes/pisces.js?v=7.1.0"></script>



  
  <script src="/js/scrollspy.js?v=7.1.0"></script>
<script src="/js/post-details.js?v=7.1.0"></script>



  


  <script src="/js/next-boot.js?v=7.1.0"></script>


  

  

  

  
  

<script src="//cdn1.lncld.net/static/js/3.11.1/av-min.js"></script>



<script src="//unpkg.com/valine/dist/Valine.min.js"></script>

<script>
  var GUEST = ['nick', 'mail', 'link'];
  var guest = 'nick,mail,link';
  guest = guest.split(',').filter(function(item) {
    return GUEST.indexOf(item) > -1;
  });
  new Valine({
    el: '#comments',
    verify: true,
    notify: false,
    appId: 'jjdDCzDO6jlN5yOgAbnVwdJ3-gzGzoHsz',
    appKey: 'jnjP7aorboCG7pJV6HvdMP15',
    placeholder: 'ヾﾉ≧∀≦)o来啊，快活啊!',
    avatar: 'mm',
    meta: guest,
    pageSize: '10' || 10,
    visitor: false,
    lang: 'zh-cn' || 'zh-cn'
  });
</script>




  


  




  

  

  

  

  

  

  

  

  

  

  

  

  

  

</body>
</html>
